package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.common.pojo.Point0F3D;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.service.LoadPoint0F3DService;
import com.navinfo.tripanalysis.offline.util.PointUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * 加载0F3D数据服务抽象类
 * @author hmc
 */
public abstract class AbstractLoadPoint0F3DService extends AbstractHiveLoadService implements LoadPoint0F3DService {

    /**
     * tid字段默认字段名
     */
    protected String TID_NAME = "terminalid";
    /**
     * 序列号字段默认字段名
     */
    protected String SERIALNUMBER_NAME = "serialnumber";
    /**
     * 是否支持serialnumber字段
     */
    protected boolean supportSerialNumber;
    /**
     * 加载的字段列表
     */
    protected static final List<String> COL_NAMES = new ArrayList<>();

    public boolean isSupportSerialNumber() {
        return supportSerialNumber;
    }

    /**
     * 是否支持序列号字段
     * @param supportSerNumber
     */
    public void setSupportSerialNumber(boolean supportSerNumber) {
        if (supportSerNumber) {
            if (!COL_NAMES.contains(SERIALNUMBER_NAME)) {
                COL_NAMES.add(SERIALNUMBER_NAME);
            }
        } else {
            if (COL_NAMES.contains(SERIALNUMBER_NAME)) {
                COL_NAMES.remove(SERIALNUMBER_NAME);
            }
        }
        this.supportSerialNumber = supportSerNumber;
    }
    @Override
    protected List<String> getSelectColumns() {
        return new ArrayList<>(COL_NAMES);
    }

    @Override
    protected List<String> getHDFSRealPaths(LoadDataParam param) {
        return PointUtils.getHDFSPathWithoutHashtid(getHdfsPath(), param);
}

    @Override
    public JavaRDD<Point0F3D> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {

        return super.loadData(spark, param).mapPartitions((FlatMapFunction<Iterator<Row>, Point0F3D>) iterator ->
                new Iterator<Point0F3D>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }
                    @Override
                    public Point0F3D next() {
                        return getPoint(iterator.next());
                    }
                });
    }

    /**
     * 将row转为具体的Point类型
     * @param row
     * @return
     */
    protected abstract Point0F3D getPoint(Row row);
}
